package com.biodiv.client.handler;


import com.biodiv.client.config.ThreadPoolConfig;
import com.biodiv.common.instance.ServiceInstance;
import com.biodiv.common.message.*;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.concurrent.ScheduledFuture;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * @Author: tianquan
 * @date: 2024-11-03  12:19
 * @Description: 客户端
 */
@Slf4j
@ChannelHandler.Sharable
public class ClientChannelInboundHandlerAdapterHandler extends ChannelInboundHandlerAdapter {

    private String username;

    private String password;

    private ServiceInstance serviceInstance;

    ScheduledFuture<?> registerFuture;

    ScheduledFuture<?> PullServiceFuture;

    private final CountDownLatch WAIT_FOR_LOGIN = new CountDownLatch(1);

    AtomicBoolean LOGIN = new AtomicBoolean(false);

    public ClientChannelInboundHandlerAdapterHandler(String username, String password, ServiceInstance serviceInstance) {
        this.username = username;
        this.password = password;
        this.serviceInstance = serviceInstance;
    }

    // 连接事件
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        PullServiceRequestMessage pullServiceRequestMessage = new PullServiceRequestMessage();
        ServiceRegistrationMessage serviceRegistrationMessage = new ServiceRegistrationMessage(serviceInstance);
        // 启动线程
        ThreadPoolConfig.THREAD_POOL.execute(() -> {
            LoginRequestMessage message = new LoginRequestMessage(username, password);
            log.info("用户名{},密码{}", username, password);
            // 发送登录消息
            ctx.writeAndFlush(message);
            // 等待登录
            try {
                WAIT_FOR_LOGIN.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            }
            // 如果登录失败
            if (!LOGIN.get()) {
                log.info("登录失败");
                ctx.channel().close();
            }
            // 定时任务5秒上报一次服务
            registerFuture = ctx.executor()
                    .scheduleAtFixedRate(() -> ctx.writeAndFlush(serviceRegistrationMessage), 0, 5, TimeUnit.SECONDS);
            // 定时任务15秒拉取一次所有服务
            PullServiceFuture = ctx.executor()
                    .scheduleAtFixedRate(() -> ctx.writeAndFlush(pullServiceRequestMessage), 1, 15, TimeUnit.SECONDS);
        });
    }

    // 可读事件
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 判断是否是登录消息
        if ((msg instanceof LoginResponseMessage)) {
            LoginResponseMessage response = (LoginResponseMessage) msg;
            if (response.isSuccess()) {
                // 如果登录成功
                LOGIN.set(true);
                log.info("登录成功");
            }
            // 唤醒Active事件
            WAIT_FOR_LOGIN.countDown();
        }
        super.channelRead(ctx, msg);
    }

    // 在连接断开时触发
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.debug("=============连接已经断开=================");
        if (registerFuture != null) {
            registerFuture.cancel(false);
        }
        if (PullServiceFuture != null) {
            PullServiceFuture.cancel(false);
        }
        // 发送消息给服务端,删除实例
        ctx.writeAndFlush(new DeleteInstanceMessage(serviceInstance.getId(), serviceInstance.getServerName()));
    }

    // 在出现异常时触发
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 发送消息给服务端,删除实例
        ctx.writeAndFlush(new DeleteInstanceMessage(serviceInstance.getId(), serviceInstance.getServerName()));
        ctx.close();
    }

}
